package cn.jiyun.kafka.stream.listener;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

/**
 * 监听类
 */
@EnableBinding(CountNumListener.CountNumProcessor.class)
public class CountNumListener {

    @StreamListener("count-num-stream-topic")
    @SendTo("count-num-stream-consumer")
    public KStream<String, String> process(KStream<String,String> input){ //c2:3
        return input.flatMap(new KeyValueMapper<String, String, Iterable<KeyValue<String, Integer>>>() {
            @Override
            public Iterable<KeyValue<String, Integer>> apply(String key, String value) {
                String[] split = value.split(":");
                List<KeyValue<String, Integer>> list = new ArrayList<>();
                list.add(new KeyValue<>(split[0], Integer.parseInt(split[1])));
                return list;
            }
        }).groupByKey(Grouped.with(Serdes.String(),Serdes.Integer())).
                windowedBy(TimeWindows.of(Duration.ofSeconds(15))).
                aggregate(()->0,(key,value,agg)->value+agg, Materialized.with(Serdes.String(),Serdes.Integer())).toStream()
                .map((key,value)->new KeyValue<>(key.key(),String.valueOf(value)));
    }



    public interface CountNumProcessor{
        // 指定输入topic
        @Input("count-num-stream-topic")
        KStream<String,String> input();
        // 指定输出topic
        @Output("count-num-stream-consumer")
        KStream<String, String> output();
    }
}
